package oc.tm.sg.core.rpc.consumer;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import oc.tm.sg.core.conf.GovernConf;
import oc.tm.sg.core.loadbalance.LoadBalance;
import oc.tm.sg.core.loadbalance.policy.RandomLoadBalance;
import oc.tm.sg.core.loadbalance.support.LoadServer;
import oc.tm.sg.core.monitor.MemoryCollectMonitor;
import oc.tm.sg.core.monitor.Monitor;
import oc.tm.sg.core.rpc.SgRpcFactory;
import oc.tm.sg.core.util.NetUtils;
import oc.tm.sg.registry.common.Constants;
import oc.tm.sg.registry.common.URL;
import oc.tm.sg.registry.common.extension.ExtensionLoader;
import oc.tm.sg.registry.core.NotifyListener;
import oc.tm.sg.registry.core.Registry;
import oc.tm.sg.registry.core.RegistryFactory;

import org.aeonbits.owner.ConfigFactory;
import org.zbus.broker.Broker;
import org.zbus.broker.BrokerConfig;
import org.zbus.broker.SingleBroker;
import org.zbus.kit.log.Logger;

/**
 * RPC消费者工厂
 * @author lry
 */
public enum RpcConsumerFactory {

	INSTANCE;
	private static final Logger logger=Logger.getLogger(RpcConsumerFactory.class); 

	private RegistryFactory factory =null;
	private Registry registry =null;

	private ConcurrentHashMap<String,List<URL>> subscribeURLMap=new ConcurrentHashMap<String, List<URL>>();
	private ConcurrentHashMap<String,NotifyListener> subscribeNotifyListenerMap=new ConcurrentHashMap<String, NotifyListener>();
	
	//远程服务缓存:如果指定服务路由至指定的地址,则回收利用之前创建过的记录,而无需重新建立服务
	private volatile transient ConcurrentHashMap<String,Object> cacheRpcServices=new ConcurrentHashMap<String,Object>();
	
	/**
	 * 配置信息
	 */
	private GovernConf governConf=null;
	
	private Broker broker=null;
	
	private Monitor monitor;
	private Monitor memoryMonitor;
	
	public GovernConf getGovernConf() {
		return governConf;
	}
	public Monitor getMemoryMonitor(){
		return memoryMonitor;
	}
	
	public void start(){
		governConf=ConfigFactory.create(GovernConf.class);

		try {
			factory = ExtensionLoader.getExtensionLoader(RegistryFactory.class).getAdaptiveExtension();
	    	registry = factory.getRegistry(URL.valueOf(governConf.registryURL()));
	    	if(registry.isAvailable()){
	    		monitor=RpcConsumerFactory.INSTANCE.getService(Monitor.class);
				memoryMonitor=new MemoryCollectMonitor(monitor);
	    	}else{
	    		if(logger.isDebugEnabled()){
					logger.debug("Registry is UnAvailable!");
				}
	    	}
		} catch (Throwable t) {
			logger.info("Started Consumer is failed!");
			t.printStackTrace();
		}
	}
	
	/**
	 * 统一调度
	 * @param clazz
	 * @return
	 */
	@SuppressWarnings("unchecked")
	public <T> T getService(Class<T> clazz) {
		//服务接口
		String key=clazz.getName();
		
		//订阅组装
		Map<String, String> parameters=new HashMap<String, String>();
		parameters.put(Constants.APPLICATION_KEY, getGovernConf().application());
		parameters.put(Constants.INTERFACE_KEY, key);
		parameters.put(Constants.GROUP_KEY, Constants.ANY_VALUE);
		parameters.put(Constants.VERSION_KEY, Constants.ANY_VALUE);
		parameters.put(Constants.CLASSIFIER_KEY, Constants.ANY_VALUE);
		parameters.put(Constants.CATEGORY_KEY, Constants.PROVIDERS_CATEGORY);
		parameters.put(Constants.CHECK_KEY, String.valueOf(false));
		final URL subscribURL = new URL(oc.tm.sg.core.Constants.DEFAULT_PROTOCOL, NetUtils.getLocalIp(), 0, "",parameters);
		
		//订阅
		List<URL> subscribeURLList=subscribeURLMap.get(subscribURL.toFullString());
		NotifyListener notifyListener=subscribeNotifyListenerMap.get(subscribURL.toFullString());
		if(subscribeURLList==null||notifyListener==null){
			//消费者注册
			registry.register(subscribURL.addParameter(Constants.CATEGORY_KEY, Constants.CONSUMERS_CATEGORY));
			
			//监听器
			NotifyListener listener=new NotifyListener() {
				public void notify(List<URL> urls) {
					List<URL> notifyURLs=new ArrayList<URL>();
					for (URL url:urls) {
						if(!Constants.EMPTY_PROTOCOL.equals(url.getProtocol())){//筛选
							notifyURLs.add(url);
						}
					}
					subscribeURLMap.put(subscribURL.toFullString(), notifyURLs);
				}
			};
			//订阅
			registry.subscribe(subscribURL, listener);
			//缓存
			subscribeNotifyListenerMap.put(subscribURL.toFullString(), listener);
			
			//重新读取
			subscribeURLList=subscribeURLMap.get(subscribURL.toFullString());
			notifyListener=subscribeNotifyListenerMap.get(subscribURL.toFullString());
		}
		
		if(subscribeURLList==null||subscribeURLList.isEmpty()){
			if(logger.isDebugEnabled()){
				logger.debug("No available service providers.");
			}
		}else{
			//初始化负载均衡数据
			List<LoadServer<URL>> loadList=new ArrayList<LoadServer<URL>>();
			for (URL url:subscribeURLList) {
				int weight=url.getParameter(Constants.WEIGHT_KEY, Constants.DEFAULT_WEIGHT);
				loadList.add(new LoadServer<URL>(url, weight));
			}
			
			//负载均衡
			URL loadURL=null;
			if(!loadList.isEmpty()){
				LoadBalance<LoadServer<URL>> loadBalance=new RandomLoadBalance();
				LoadServer<URL> loadServer=loadBalance.select(loadList);
				
				if(loadServer!=null){
					loadURL=loadServer.getServer();
				}
			}
			
			//消费
			if(loadURL!=null){
				try {
					if(logger.isInfoEnabled()){
						logger.info("Invoker url is: "+loadURL.toFullString());
					}
					
					//缓存检查
					String interfaceAddress=key+"@"+loadURL.getAddress();
					Object service=cacheRpcServices.get(interfaceAddress);
					if(service!=null){
						return (T) service;
					}else{
						//创建服务
						T newService=getService(loadURL.getAddress(),loadURL.getParameters(), clazz);	
						//记录缓存
						cacheRpcServices.put(interfaceAddress, newService);
						return newService;
					}
				} catch (Throwable t) {
					logger.error("Invoker failure.");
					t.printStackTrace();
				}
			}else{
				if(logger.isDebugEnabled()){
					logger.debug("Load failed.");
				}
			}
		}
		return null;
	}
	
	
	/**
	 * 启动服务提供者
	 */
	public <T> T getService(String address,Map<String,String> param,Class<T> clazz) {
		BrokerConfig brokerConfig = new BrokerConfig(); 
		brokerConfig.setBrokerAddress(address);
		return getService(brokerConfig,param,clazz);
	}
	
	/**
	 * 启动服务提供者
	 * @param <T>
	 * @param config
	 */
	public <T> T getService(BrokerConfig brokerConfig,Map<String,String> param,Class<T> clazz) {
		if(logger.isInfoEnabled()){
			logger.info("Consumer is invoking,please waitting...");
		}
		
		try {
			broker = new SingleBroker(brokerConfig); 
			SgRpcFactory factory = new SgRpcFactory(brokerConfig,broker);
			T result = factory.getService(clazz,this,param);
			
			if(logger.isInfoEnabled()){
				logger.info("Consumer invoked is successed!");
			}
			return result;
		} catch (Throwable t) {
			logger.info("Consumer invoked is failed!");
			t.printStackTrace();
		}
		return null;
	}
	

	/**
	 * 销毁
	 */
	public void destroy() {
		if(logger.isInfoEnabled()){
			logger.info("Closing Consumer, please waitting...");
		}
		try{
			if(broker!=null){
				broker.close();
			}
			if(logger.isInfoEnabled()){
				logger.info("Closed Consumer is successed!");
			}
		} catch (Throwable t) {
			logger.info("Closed Consumer is failed!");
			t.printStackTrace();
		}
	}

}
